package com.tech.consumer

import com.tech.common.KuduUtil
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.StreamingQueryListener

class MyStreamingQueryListener(val spark: SparkSession, val loadResourceManager: LoadResourceManager) extends StreamingQueryListener {
  override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit = {
    //    println("Query started:" + event.id)
  }

  override def onQueryProgress(event: StreamingQueryListener.QueryProgressEvent): Unit = {
    //    println("Query made progress: " + event.progress)

    //    val connection: Connection = DriverManager.getConnection(GlobalValue.url_mysql)
    //    if (loadResourceManager.activityBroadcast.value != MysqlUtil.getActivity(connection)) {
    //      println("开始重载广播变量")
    //      this.loadResourceManager.unPersist()
    //      this.loadResourceManager.load(spark)
    //      println("结束重载广播变量")
    //    }
    //    connection.close()
    this.synchronized {
      if (loadResourceManager.activityBroadcast.value != KuduUtil.getActivity4D365) {
        println("开始重载活动规则")
        this.loadResourceManager.unPersist()
        this.loadResourceManager.load4D365(spark)
        println("结束重载活动规则")
      }
      if (loadResourceManager.dimDataGroupDetailMapBroadcast.value != KuduUtil.getDimDataGroupDetail){
        println("开始重载门店及商品信息")
        this.loadResourceManager.unPersistDimDataGroupDetail()
        this.loadResourceManager.loadDimDataGroupDetail(spark)
        println("结束重载门店及商品信息")
      }
    }


  }

  override def onQueryTerminated(event: StreamingQueryListener.QueryTerminatedEvent): Unit = {
    //    println("Query terminated: " + event.id)
  }
}
